package com.zhang.hadoop.flink.test9;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;

/**
 * @author: zhang yufei
 * @createTime:2022/11/5 11:26
 * @description:
 */
public class UdfTest_AggregateFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //`1,在创建表的DDL中直接定义时间属性
        String createDDL = "create table clickTable(" +
                " user_name string," +
                " url string," +
                " ts bigint," +
                " et as to_timestamp(from_unixtime(ts/1000))," +
                " watermark for et as et - interval '1' second " +
                ") with (" +
                " 'connector' = 'filesystem'," +
                " 'path'='flink/input/clicks.txt'," +
                " 'format'='csv'" +
                ")";
        tableEnv.executeSql(createDDL);

        //2.注册自定义标量函数
        tableEnv.createTemporarySystemFunction("WeightedAverage", WeightedAverage.class);

        //3.调用UDF进行查询转换
        Table resultTable = tableEnv.sqlQuery(
                "select user_name,WeightedAverage(ts,1) as w_avg" +
                        " from clickTable group by user_name");

        //4.转换成流打印输出
        tableEnv.toChangelogStream(resultTable).print();

        env.execute();
    }

    //单独定义一个累加器类型
    public static class WeightAvgAccumulator {

        public long sum = 0;

        public int count = 0;
    }

    //实现自定义的聚合函数，计算加权平均值
    public static class WeightedAverage extends AggregateFunction<Long, WeightAvgAccumulator> {

        @Override
        public Long getValue(WeightAvgAccumulator accumulator) {
            if (accumulator.count == 0) {
                return null;
            } else {
                return accumulator.sum / accumulator.count;
            }
        }

        @Override
        public WeightAvgAccumulator createAccumulator() {
            return new WeightAvgAccumulator();
        }

        //累加计算的方法
        public void accumulate(WeightAvgAccumulator accumulator, Long iValue, Integer iWeight) {
            accumulator.sum = accumulator.sum + (iValue * iWeight);
            accumulator.count = accumulator.count + iWeight;
        }
    }
}
